package com.ibm.cps.kafka;

import org.apache.log4j.Logger;

import com.ibm.util.exception.CPSException;

import kafka.consumer.ConsumerIterator;

public abstract class KafkaTopicConsumeTemplate {

	private Logger logger;
	private ConsumerIterator<byte[], byte[]> it;

	public KafkaTopicConsumeTemplate(String zookeeper, String topicName) {
		this(zookeeper, "groupID", topicName);
	}

	public KafkaTopicConsumeTemplate(String zookeeper, String groupid,
			String topicName) {
		logger = Logger.getLogger(KafkaTopicConsumeTemplate.class);
		// KafkaTopicUtils.createTopic(zookeeper, topicName, 1, 1);
		logger.info("Creating connection to kafka topic: " + topicName);
		KafkaConsumer consumer = new KafkaConsumer(zookeeper, groupid);
		logger.info("Connected to kafka topic: " + topicName);
		it = consumer.consumeTopic(topicName).iterator();
		logger.info("Waiting for message from topic " + topicName
				+ " in zookeeper " + zookeeper + "...");
	}

	public void consumeMessage() throws CPSException {
		if (it.hasNext()) {
			byte[] byteMessage = it.next().message();
			process(byteMessage);
		}
	}

	public void consumeMessages() throws CPSException {
		while (it.hasNext()) {
			byte[] byteMessage = it.next().message();
			process(byteMessage);
		}
	}

	public abstract void process(byte[] message) throws CPSException;

}
